#include <memory>
#include <iostream>
#include <brpc/server.h>
#include <butil/logging.h>

// common文件夹下
#include "mysql_chat_session_member.hpp"
#include "etcder.hpp"
#include "logger.hpp"
#include "channel.hpp"
#include "uuid.hpp"
#include "rabbitmq.hpp"
#include "user.pb.h"
#include "base.pb.h"
#include "transmit.pb.h"

namespace bite_im
{
    // 实现消息转发功能

    // 消息转发功能服务类
    class MsgTransmitServiceImpl : public bite_im::MsgTransmitService
    {
    public:
        using ptr = std::shared_ptr<MsgTransmitServiceImpl>;
        MsgTransmitServiceImpl(const DBPtr &mysql_client,
                               const MQClient::ptr &mq_client,
                               const ServiceManager::ptr &service_manager,
                               const std::string &user_service_name,
                               const std::string &exchange_name)

            : _mysql_session_member(std::make_shared<ChatSessionMemberTable>(mysql_client)),
              _mq_client(mq_client),
              _service_manager(service_manager),
              _user_service_name(user_service_name),
              _exchange_name(exchange_name)
        {
        }
        ~MsgTransmitServiceImpl() {}
        void GetTransmitTarget(google::protobuf::RpcController *controller,
                               const ::bite_im::NewMessageReq *request,
                               ::bite_im::GetTransmitTargetRsp *response,
                               ::google::protobuf::Closure *done)
        {
            /*
                场景：前端准备好发送的信息结构和聊天会话id，发送给网关填入用户信息，接着网关通过
                    消息转发子服务的GetTransmitTarget获得一个转发列表，并将这个消息放到消息队列mq里面,
                    与此同时消息存储子服务通过回调函数，在这个mq队列中消费这条消息，存储这条消息。
                    并且网关也向当前聊天会话内的所有成员发起新消息的通知，消息内容伴随这个通知也推送过去
            */

            // 错误函数---lambda
            auto err_response_func = [this, response](const std::string &request_id,
                                                      const std::string &errmsg) -> void
            {
                response->set_request_id(request_id);
                response->set_success(false);
                response->set_errmsg(errmsg);
                return;
            };
            // 管理rpc对象
            brpc::ClosureGuard rpc_guard(done);

            // 1.从请求中获取发送信息的用户id，所属会话id，消息内容
            std::string user_id = request->user_id();
            std::string chat_session_id = request->chat_session_id();
            // const MessageContent &content = request->message();

            // 2.构建用户子服务
            auto channel = _service_manager->choose(_user_service_name);
            if (!channel) // 服务不存在
            {
                LOG_ERROR("用户子服务[{}]节点未找到，请求ID[{}]", _user_service_name, request->request_id());
                return err_response_func(request->request_id(), "未找到用户子服务管理节点");
            }

            UserService_Stub stub(channel.get());

            // 3.获取消息发送的发送者用户信息
            GetUserInfoReq req;
            req.set_request_id(request->request_id());
            req.set_user_id(request->user_id());
            GetUserInfoRsp resp;

            brpc::Controller cntl;
            stub.GetUserInfo(&cntl, &req, &resp, nullptr);
            if (cntl.Failed() == true || resp.success() == false)
            {
                LOG_ERROR("用户子服务调用失败，请求ID[{}]，原因[{}]", request->request_id(), cntl.ErrorText());
                return err_response_func(request->request_id(), "用户子服务调用失败");
            }
            UserInfo user_info = resp.user_info(); // 发送者用户信息
            
            // 组织消息结构---谁发送的、发送了什么
            MessageInfo message_info;
            message_info.set_message_id(UUID::uuid());
            message_info.set_chat_session_id(chat_session_id);
            message_info.set_timestamp(time(nullptr));
            message_info.mutable_sender()->CopyFrom(resp.user_info());
            message_info.mutable_message()->CopyFrom(request->message());

            // 5.获取会话中的成员user_id列表
            auto user_id_list = _mysql_session_member->members(chat_session_id);

            // 6.将信息发送到消息队列中，内部实现的routing_key为宏default_routing_key
            bool ret = _mq_client->publish(_exchange_name, message_info.SerializeAsString()); 
            if (ret == false)
            {
                LOG_ERROR("消息队列发布消息失败，请求ID[{}]", request->request_id());
                return err_response_func(request->request_id(), "消息队列发布消息");
            }
            // 7.组织响应
            response->set_request_id(request->request_id());
            response->set_success(true);
            response->mutable_message()->CopyFrom(message_info);
            for (auto target_id : user_id_list)
            {
                response->add_target_id_list(target_id);
            }
        }

    private:
        // 用户调用子服务
        std::string _user_service_name;
        ServiceManager::ptr _service_manager;
        // 会话成员表
        ChatSessionMemberTable::ptr _mysql_session_member;
        // rabbitmq客户端
        std::string _exchange_name;
        MQClient::ptr _mq_client;
    };

    class MsgTransmitServer
    {
        using RpcServerPtr = std::shared_ptr<brpc::Server>;

    public:
        using ptr = std::shared_ptr<MsgTransmitServer>;
        MsgTransmitServer(const Discovery::ptr &dis_client,
                          const Registry::ptr &reg_client,
                          const DBPtr &mysql_client,
                          const RpcServerPtr &rpc_server)
            : _dis_client(dis_client),
              _reg_client(reg_client),
              _mysql_client(mysql_client),
              _rpc_server(rpc_server) {}

        ~MsgTransmitServer() {}

        // 启动RPC服务器
        void start()
        {
            _rpc_server->RunUntilAskedToQuit();
        }

    private:
        Discovery::ptr _dis_client;
        Registry::ptr _reg_client;
        DBPtr _mysql_client;
        RpcServerPtr _rpc_server;
    };

    /*建造者设计模式，封装对象的构造过程，防止用户在进行操作时，对未构造的对象进行操作*/

    class MsgTransmitServerBuilder
    {
        using RpcServerPtr = std::shared_ptr<brpc::Server>;

    public:
        void make_mq_object(const std::string &mq_user_name,
                            const std::string &mq_password,
                            const std::string &mq_host,
                            const std::string &exchange_name,
                            const std::string &queue_name,
                            const std::string &binding_key)
        {
            _mq_client = std::make_shared<MQClient>(mq_user_name, mq_password, mq_host);
            _mq_client->declareComponents(exchange_name, queue_name, binding_key);  // binding_key需要跟MQClient中的routing_key一致
            _exchange_name = exchange_name;
        }
        void make_mysql_object(const std::string &mysql_user_name,
                               const std::string &mysql_password,
                               const std::string &db_name,
                               const std::string &mysql_host,
                               int mysql_port,
                               const std::string &cset,
                               int conn_pool_num)
        {
            _mysql_client = ODBFactory::create(mysql_user_name, mysql_password, db_name,
                                               mysql_host, mysql_port, cset, conn_pool_num);
        }
        // 构造服务发现和服务管理客户端
        void make_find_object(const std::string &reg_host,
                              const std::string &base_service_name,
                              const std::string &user_service_name)
        {
            _user_service_name = user_service_name;
            _service_manager = std::make_shared<ServiceManager>();
            _service_manager->declareService(user_service_name);
            LOG_INFO("在用户服务中添加了用户管理子服务[{}]", user_service_name);
            auto put_cb = std::bind(&bite_im::ServiceManager::onServiceOnline, _service_manager.get(),
                                    std::placeholders::_1, std::placeholders::_2);
            auto del_cb = std::bind(&bite_im::ServiceManager::onServiceOffLine, _service_manager.get(),
                                    std::placeholders::_1, std::placeholders::_2);
            _dis_client = std::make_shared<Discovery>(base_service_name, reg_host, put_cb, del_cb);
        }
        // 构造服务注册客户端对象
        void make_reg_object(const std::string &reg_host,
                             const std::string &service_name,
                             const std::string &access_host)
        {
            _reg_client = std::make_shared<Registry>(reg_host);
            _reg_client->registry(service_name, access_host);
        }
        // 搭建rpc服务器
        void make_rpc_server(uint16_t port, int32_t time_out, uint8_t num_threads)
        {
            // 差错检测
            if (_mq_client.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未初始化消息队列模块");
                abort();
            }
            if (_mysql_client.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未初始化mysql客户端模块");
                abort();
            }
            if (_service_manager.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未初始化服务管理模块");
                abort();
            }
            // 创建rpc服务器对象
            _rpc_server = std::make_shared<brpc::Server>();
            // 构建服务，并添加服务
            MsgTransmitServiceImpl *transmit_service = new MsgTransmitServiceImpl(_mysql_client, _mq_client, _service_manager,
                                                                                  _user_service_name, _exchange_name); // 注意这里需要在堆上开辟空间
            int ret = _rpc_server->AddService(transmit_service, brpc::ServiceOwnership::SERVER_OWNS_SERVICE);
            if (ret == -1)
            {
                LOG_ERROR("rpc服务器添加服务失败");
                abort();
            }

            auto options = std::make_shared<brpc::ServerOptions>();
            // 启动服务器
            options->idle_timeout_sec = time_out; // 连接空闲超时时间
            options->num_threads = num_threads;
            ret = _rpc_server->Start(port, options.get());
            if (ret == -1)
            {
                LOG_ERROR("服务器启动失败");
                abort();
            }
        }
        MsgTransmitServer::ptr build()
        {
            // 检测维护对象是否建立
            if (_dis_client.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未初始化服务发现模块");
                abort();
            }
            if (_reg_client.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未初始化服务注册模块");
                abort();
            }
            if (_rpc_server.get() == nullptr)
            {
                LOG_ERROR("服务器建造失败，未搭建Rpc服务器");
                abort();
            }
            MsgTransmitServer::ptr server = std::make_shared<MsgTransmitServer>(_dis_client, _reg_client, _mysql_client, _rpc_server);
            return server;
        }

    private:
        // 消息队列
        MQClient::ptr _mq_client;
        std::string _exchange_name;
        // 数据库
        DBPtr _mysql_client;
        // 服务注册和发现模块
        std::string _user_service_name;
        ServiceManager::ptr _service_manager;
        Discovery::ptr _dis_client;
        Registry::ptr _reg_client;

        RpcServerPtr _rpc_server;
    };
}